/*
	发布者：
		1. 发布信息到channel
		2. 维护topic
*/
package amq

import (
	"log"
	"sync"
	"time"
)

type subChannel chan interface{}

type Publisher struct {
	Index   string        // 索引名称
	buffer  int           // 缓存区大小
	timeout time.Duration // 超时时间
	subMap  sync.Map      // 并发map[string][]subChannel
}

// 新建发布者
func NewPublisher(index string, buffer int, timeout time.Duration) *Publisher {
	return &Publisher{
		Index:   index,
		buffer:  buffer,
		timeout: timeout,
	}
}

// 在Topic下注册通道
func (p *Publisher) PublishTopic(topic string, channel subChannel) {
	var subList []subChannel
	load, _ := p.subMap.LoadOrStore(topic, subList)
	subList = load.([]subChannel)
	subList = append(subList, channel)
	p.subMap.Store(topic, subList) // 重新存入不然无效
}

// 推送消息
func (p *Publisher) SendMsg(topic string, msg interface{}) {
	load, ok := p.subMap.Load(topic)
	if !ok {
		log.Println("推送失败，暂无该topic -> ", topic)
	}
	subList := load.([]subChannel)
	go p.pushMsg(subList, msg)
}

// 发送到订阅的手中
func (p *Publisher) pushMsg(subList []subChannel, msg interface{}) {
	for _, sub := range subList {
		select {
		case sub <- msg:
		case <-time.After(p.timeout):
		}
	}
}
